package gbench.appdemo.myworld.economy;

import static gbench.common.tree.LittleTree.IRecord.REC;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import gbench.common.tree.LittleTree.IRecord;

/**
 * 经济体：所谓经济体 就是 一个拥有 自己自足能力 的 生产消费 活动的联合体 <br>
 * 
 * Economy 经济用于 表达需求 <br>
 * Market 用于实现求
 * @author gbench
 *
 */
public class Economy {

    /**
     * 构造一个经济体
     * @param buffersize 经济体的消息buffer 容量
     */
    public Economy() {
        this(1024);
    }

    /**
     * 构造一个经济体
     * @param buffersize 经济体的消息buffer 容量
     */
    public Economy(final int buffersize) {
        this.bufferSize = buffersize;
        this.disruptor = new Disruptor<IRecord>(() -> REC(), bufferSize, Executors.defaultThreadFactory(),
                ProducerType.MULTI, new YieldingWaitStrategy());
    }

    /**
     * 添加一种消费活动 
     * @param handlers 消费活动 列表
     * @return 经济体对象 用于实现链式编程
     */
    @SafeVarargs
    public final synchronized Economy consume(EventHandler<? super IRecord>... handlers) {
        this.disruptor.handleEventsWith(handlers);
        return this;
    }

    /**
     * 同步运行
     * @param cs 增加一个 消费活动
     * @return 经济体对象 用于实现链式编程
     */
    public final synchronized Economy consume(final Consumer<IRecord> cs) {
        this.activities.add(CACT(eco -> {
            eco.disruptor.handleEventsWith((event, seq, batch) -> {
                cs.accept(event.derive(REC("batch", batch)));// 使用derive 衍生出一个 event 结构 可以防止 多 Consumer 并发时候的 ConcurrentModificationException
            });
        }));
        return this;
    }

    /**
     * 提交消息时间:向经济体提供一项 产品消息
     * 
     * @param rec 消息记录
     */
    public synchronized void publish(final IRecord rec) {
        disruptor.getRingBuffer().publishEvent((r, seq) -> r.add(rec).add("seq", seq));
    }

    /**
     * 同步运行： <br>
     * 向经济体提供一项 产品消息
     * @param cs 消息发布函数
     */
    public synchronized Economy produce(Consumer<Economy> cs) {
        this.activities.add(PACT(cs));
        return this;
    }
    
    /**
     * 向经济体提供一项 产品消息
     * @param supplier 消息供应函数
     */
    public synchronized Economy produce(Supplier<List<IRecord>>supplier) {
        return this.produce(eco->supplier.get().stream().forEach(rec->eco.publish(rec)));
    }

    /**
     * 启动经济体 <br>
     * 开始运行:异步云信接口
     *  
     * @param finisher,finisher 需要设置 卡点
     */
    public synchronized Economy collect(Consumer<Economy> finisher) {
        new Thread(() -> {
            final var registry = this.activities.stream().collect(Collectors.groupingBy(a->a.type));// 回调函数注册表
            registry.computeIfPresent(ActivityType.CONSUME, (k,actions)->{ // 消费函数注册表
                actions.forEach(action->action.handler.accept(this));
                return actions;
            });
            disruptor.start();
            registry.computeIfPresent(ActivityType.PRODUCE, (k,actions)->{ // 生产函数注册表
                // 异步执行生产者
                new Thread(()->actions.parallelStream().forEach(action->action.handler.accept(this))).start();;
                return actions;
            });
            finisher.accept(this);
        }).start();
        return this;
    }
    
    /**
     * 启动disruptor 
     */
    public void start() {
        this.disruptor.start();
    }

    /**
     * 关闭 disruptor
     */
    public void shutdown() {
        this.disruptor.shutdown();
    }
    
    /**
     * 经济活动：生产活动与消费活动
     * @author gbench
     *
     */
    public class Activity{
        
        /**
         * 构造函数
         * @param type
         * @param handler
         */
        public Activity(ActivityType type,Consumer<Economy> handler){
            this.type = type;
            this.handler = handler;
        }
        
        ActivityType type;
        Consumer<Economy> handler;
    }
    
    /**
     * 生产活动
     * @param cs 生产活动  economy->{ 发送消息 }
     * @return 经济活动
     */
    public Activity PACT(Consumer<Economy> cs) {
        return new Activity(ActivityType.PRODUCE,cs);
    }
    
    /**
     * 消费活动 
     * @param cs 消费活动  record->{ 处理消息 }
     * @return 消费活动
     */
    public Activity CACT(Consumer<Economy> cs) {
        return new Activity(ActivityType.CONSUME,cs);
    }

    final int bufferSize;
    final Disruptor<IRecord> disruptor;
    final List<Activity> activities = new LinkedList<>();
    public enum ActivityType{CONSUME,PRODUCE};// 顺序很重要，一定要CONSUME在前
}